nanopyx.methods.workflow

View Source

 1from ..__agent__ import Agent
 2
 3class Workflow:
 4    
 5    """ 
 6    Workflow class that aggregates all the steps of the analysis and the corresponding functions.
 7    """
 8
 9    def __init__(self,*args) -> None:
10        """
11        Each arg is a tuple of 3 items (fn, args, kwargs)
12        """
13
14        self._methods = []
15        self._return_values = []
16
17        for arg in args:
18            if isinstance(arg, tuple) and len(arg) == 3:
19                self._methods.append((arg[0], arg[1], arg[2]))
20            else:
21                raise TypeError("Each arg must be a tuple of 3 items (fn, args, kwargs)")
22
23    def run(self,_force_run_type=None):
24        """
25        Run the workflow sequentially
26        """
27
28        for method in self._methods:
29            fn, args, kwargs = method
30
31            # in the list args, substitute 'PREV_RETURN_VALUE' with the return value of the previous method
32            sane_args = []
33            for arg in args:
34                if isinstance(arg, str) and 'PREV_RETURN_VALUE' in arg:
35                    indices = [int(i) for i in arg.split('_') if i.isdigit()]
36                    return_value = self._return_values[indices[0]][indices[1]]
37                    sane_args.append(return_value)
38                else:
39                    sane_args.append(arg)
40
41            # in the dict kwargs, substitute 'PREV_RETURN_VALUE' with the return value of the previous method
42            for key, value in kwargs.items():
43                if isinstance(value, str) and 'PREV_RETURN_VALUE' in value:
44                    indices = [int(i) for i in value.split('_') if i.isdigit()]
45                    return_value = self._return_values[indices[0]][indices[1]]
46                    kwargs[key] = return_value
47
48
49            # Get run type from the Agent
50            run_type = Agent.get_run_type(fn, sane_args, kwargs)
51            kwargs['run_type'] = run_type
52
53            if _force_run_type:
54                kwargs['run_type'] = _force_run_type
55
56            # TODO maybe we need to warn the agent its running and when it finishes
57            return_value = fn.run(*sane_args, **kwargs)
58
59            if isinstance(return_value, tuple):
60                self._return_values.append(return_value)
61            else:
62                self._return_values.append((return_value,))
63
64            Agent._inform(fn)
65
66        return self._return_values[-1], fn._last_runtype, fn._last_time
67    
68    def calculate(self, _force_run_type=None):
69        output, tmp, tmp = self.run(_force_run_type=_force_run_type)
70        return output

class Workflow: View Source

 4class Workflow:
 5    
 6    """ 
 7    Workflow class that aggregates all the steps of the analysis and the corresponding functions.
 8    """
 9
10    def __init__(self,*args) -> None:
11        """
12        Each arg is a tuple of 3 items (fn, args, kwargs)
13        """
14
15        self._methods = []
16        self._return_values = []
17
18        for arg in args:
19            if isinstance(arg, tuple) and len(arg) == 3:
20                self._methods.append((arg[0], arg[1], arg[2]))
21            else:
22                raise TypeError("Each arg must be a tuple of 3 items (fn, args, kwargs)")
23
24    def run(self,_force_run_type=None):
25        """
26        Run the workflow sequentially
27        """
28
29        for method in self._methods:
30            fn, args, kwargs = method
31
32            # in the list args, substitute 'PREV_RETURN_VALUE' with the return value of the previous method
33            sane_args = []
34            for arg in args:
35                if isinstance(arg, str) and 'PREV_RETURN_VALUE' in arg:
36                    indices = [int(i) for i in arg.split('_') if i.isdigit()]
37                    return_value = self._return_values[indices[0]][indices[1]]
38                    sane_args.append(return_value)
39                else:
40                    sane_args.append(arg)
41
42            # in the dict kwargs, substitute 'PREV_RETURN_VALUE' with the return value of the previous method
43            for key, value in kwargs.items():
44                if isinstance(value, str) and 'PREV_RETURN_VALUE' in value:
45                    indices = [int(i) for i in value.split('_') if i.isdigit()]
46                    return_value = self._return_values[indices[0]][indices[1]]
47                    kwargs[key] = return_value
48
49
50            # Get run type from the Agent
51            run_type = Agent.get_run_type(fn, sane_args, kwargs)
52            kwargs['run_type'] = run_type
53
54            if _force_run_type:
55                kwargs['run_type'] = _force_run_type
56
57            # TODO maybe we need to warn the agent its running and when it finishes
58            return_value = fn.run(*sane_args, **kwargs)
59
60            if isinstance(return_value, tuple):
61                self._return_values.append(return_value)
62            else:
63                self._return_values.append((return_value,))
64
65            Agent._inform(fn)
66
67        return self._return_values[-1], fn._last_runtype, fn._last_time
68    
69    def calculate(self, _force_run_type=None):
70        output, tmp, tmp = self.run(_force_run_type=_force_run_type)
71        return output

Workflow class that aggregates all the steps of the analysis and the corresponding functions.

Workflow(*args) View Source

10    def __init__(self,*args) -> None:
11        """
12        Each arg is a tuple of 3 items (fn, args, kwargs)
13        """
14
15        self._methods = []
16        self._return_values = []
17
18        for arg in args:
19            if isinstance(arg, tuple) and len(arg) == 3:
20                self._methods.append((arg[0], arg[1], arg[2]))
21            else:
22                raise TypeError("Each arg must be a tuple of 3 items (fn, args, kwargs)")

Each arg is a tuple of 3 items (fn, args, kwargs)

def run(self, _force_run_type=None): View Source

24    def run(self,_force_run_type=None):
25        """
26        Run the workflow sequentially
27        """
28
29        for method in self._methods:
30            fn, args, kwargs = method
31
32            # in the list args, substitute 'PREV_RETURN_VALUE' with the return value of the previous method
33            sane_args = []
34            for arg in args:
35                if isinstance(arg, str) and 'PREV_RETURN_VALUE' in arg:
36                    indices = [int(i) for i in arg.split('_') if i.isdigit()]
37                    return_value = self._return_values[indices[0]][indices[1]]
38                    sane_args.append(return_value)
39                else:
40                    sane_args.append(arg)
41
42            # in the dict kwargs, substitute 'PREV_RETURN_VALUE' with the return value of the previous method
43            for key, value in kwargs.items():
44                if isinstance(value, str) and 'PREV_RETURN_VALUE' in value:
45                    indices = [int(i) for i in value.split('_') if i.isdigit()]
46                    return_value = self._return_values[indices[0]][indices[1]]
47                    kwargs[key] = return_value
48
49
50            # Get run type from the Agent
51            run_type = Agent.get_run_type(fn, sane_args, kwargs)
52            kwargs['run_type'] = run_type
53
54            if _force_run_type:
55                kwargs['run_type'] = _force_run_type
56
57            # TODO maybe we need to warn the agent its running and when it finishes
58            return_value = fn.run(*sane_args, **kwargs)
59
60            if isinstance(return_value, tuple):
61                self._return_values.append(return_value)
62            else:
63                self._return_values.append((return_value,))
64
65            Agent._inform(fn)
66
67        return self._return_values[-1], fn._last_runtype, fn._last_time

Run the workflow sequentially

def calculate(self, _force_run_type=None): View Source

69    def calculate(self, _force_run_type=None):
70        output, tmp, tmp = self.run(_force_run_type=_force_run_type)
71        return output